#include "rpcprovider.h"
#include "mprpcapplication.h"
#include <string>
#include <functional>
#include <google/protobuf/descriptor.h>
#include <rpcheader.pb.h>
#include <logger.h>
#include "zookeeperutil.h"

void RpcProvider::NotifyService(google::protobuf::Service *service)
{
    ServiceInfo serviceInfo;
    // 1.取出service服务的描述，并存入service_info中的service中

    const google::protobuf::ServiceDescriptor *serviceDesc = service->GetDescriptor();
    std::string serviceName = serviceDesc->name();
    // std::cout << "service name : " << serviceName << std::endl;
    LOG_INFO("service name: %s", serviceName.c_str());

    // 2.逐个取出service服务的方法，将方法存入service_info的方法表中
    int methodCnt = serviceDesc->method_count();
    for (int i = 0; i < methodCnt; i++)
    {
        const google::protobuf::MethodDescriptor *methodDesc = serviceDesc->method(i);
        std::string methodName = methodDesc->name();
        serviceInfo.m_methodMap.insert({methodName, methodDesc});
        // std::cout << "serviceMethod[" << i << "] name : " << methodName << std::endl;
        LOG_INFO("serviceMethod[%d] name: %s", i, methodName.c_str());
    }
    serviceInfo.m_service = service;

    // 3.service_info插入到服务表中
    m_serviceInfoMap.insert({serviceName, serviceInfo});
}

void RpcProvider::Run()
{
    std::string ip = MprpcApplication::getInstance().getConfig().load("rpcserverip");
    int port = atoi(MprpcApplication::getInstance().getConfig().load("rpcserverport").c_str());
    muduo::net::InetAddress listenAddr(ip, port);
    muduo::net::TcpServer server(&ev, listenAddr, "rpcServser");
    server.setThreadNum(4);
    server.setConnectionCallback(std::bind(&RpcProvider::onConnection, this, std::placeholders::_1));
    server.setMessageCallback(std::bind(&RpcProvider::onMessage, this, std::placeholders::_1, std::placeholders::_2, std::placeholders::_3));

    ZkClient zkCli;
    zkCli.start();
    for (auto &sp : m_serviceInfoMap)
    {
        std::string path = "/" + sp.first;
        zkCli.create(path.c_str(), nullptr, 0);
        for (auto &mp : sp.second.m_methodMap)
        {
            std::string methodPath = path + "/" + mp.first;
            char method_path_data[128] = {0};
            sprintf(method_path_data, "%s:%d", ip.c_str(), port);
            zkCli.create(methodPath.c_str(), method_path_data, strlen(method_path_data), ZOO_EPHEMERAL);
        }
    }

    // std::cout << "RpcProvider start service at " << ip << " : " << port << std::endl;
    LOG_INFO("RpcProvider start service at %s : %d", ip.c_str(), port);
    server.start();
    ev.loop();
}

void RpcProvider::onConnection(const muduo::net::TcpConnectionPtr &conn)
{
    if (!conn->connected())
    {
        conn->shutdown();
    }
}
void RpcProvider::onMessage(const muduo::net::TcpConnectionPtr &conn,
                            muduo::net::Buffer *buffer,
                            muduo::Timestamp)
{
    //请求格式 :         headr_size(4Byte) headr_str(headr_size Byte) args_str
    //其中head_str:     service_name method_name args_size
    // 1.读取字符流
    std::string buf = buffer->retrieveAllAsString(); //这里tpc粘包处理的不好
    uint32_t header_size;
    buf.copy((char *)&header_size, 4, 0);

    // 2.读取前四个字节的内容，标识请求头部长度
    std::string RpcHeader_str = buf.substr(4, header_size);

    // 3.读取请求头部，protobuf解析
    rpchaeder::RpcHaeder rpcHeader;
    std::string service_name, method_name;
    uint32_t args_size;
    if (rpcHeader.ParseFromString(RpcHeader_str))
    {
        //解析成功
        service_name = rpcHeader.service_name();
        method_name = rpcHeader.method_name();
        args_size = rpcHeader.args_size();
    }
    else
    {
        //解析失败
        // std::cout << "parse " << RpcHeader_str << " error!" << std::endl;
        LOG_ERR("parse %s error!", RpcHeader_str.c_str());
        return;
    }
    // 4.取出args
    std::string args_str = buf.substr(4 + header_size, args_size);

    //调试信息
    std::cout << "======================================" << std::endl;
    std::cout << "header_size : " << header_size << std::endl;
    std::cout << "RpcHeader_str : " << RpcHeader_str << std::endl;
    std::cout << "service_name : " << service_name << std::endl;
    std::cout << "method_name : " << method_name << std::endl;
    std::cout << "args_size : " << args_size << std::endl;
    std::cout << "======================================" << std::endl;
    // 5.查询本地的ServiceInfo表中是否有对应的服务和方法
    auto it = m_serviceInfoMap.find(service_name);
    if (it == m_serviceInfoMap.end())
    {
        // std::cout << "service : " << service_name << " is not exist!" << std::endl;
        LOG_ERR("service: %s is not exist!", service_name.c_str());
        return;
    }

    auto mit = it->second.m_methodMap.find(method_name);
    if (mit == it->second.m_methodMap.end())
    {
        // std::cout << "service : " << service_name << " method : " << method_name << " is not exist!" << std::endl;
        LOG_ERR("service: %s method: %s is not exist!", service_name.c_str(), method_name.c_str());
        return;
    }

    google::protobuf::Service *service = it->second.m_service;
    const google::protobuf::MethodDescriptor *method = mit->second;

    // 6.解析args,通过getprototype方法获得消息的原始类型,并调用new方法获得消息对象
    google::protobuf::Message *reuqest = service->GetRequestPrototype(method).New();
    if (!reuqest->ParseFromString(args_str))
    {
        // std::cout << "parse request error!" << std::endl;
        LOG_ERR("parse request error!");
        return;
    }
    google::protobuf::Message *response = service->GetResponsePrototype(method).New();
    // 7.绑定done回调函数,实现业务完成之后的序列化和网络通讯
    // Class* object, void (Class::*method)(Arg1, Arg2), Arg1 arg1, Arg2 arg2
    google::protobuf::Closure *done = google::protobuf::NewCallback<RpcProvider, const muduo::net::TcpConnectionPtr &, google::protobuf::Message *>(this, &RpcProvider::done, conn, response);
    // 8.调用rpc服务
    service->CallMethod(method, nullptr, reuqest, response, done);
}

void RpcProvider::done(const muduo::net::TcpConnectionPtr &conn, google::protobuf::Message *response)
{
    // 格式化字符串并发送
    std::string str;
    if (response->SerializeToString(&str))
    {
        conn->send(str);
    }
    else
    {
        // std::cout << "serialize respon_str error!" << std::endl;
        LOG_ERR("serialize respon_str error!");
    }

    conn->shutdown(); //模拟http短连接，由服务端主动断开连接
}
